package com.sub.kafka.producer;

import com.sub.kafka.producer.interceptor.SubProducerValueInterceptor;
import com.sub.kafka.producer.partitioner.SubPartitioner;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @ClassName SubKafkaProducer
 * @Description: Kafka生产者学习案例
 * 主要学习内容：
 * 1、生产者创建、配置及消息发送
 * 2、生产者部分配置参数
 * 3、生产者消息拦截器
 * 4、生产者分区器
 * 5、消息可靠性
 * 5.1、生产者ACK确认机制
 * 5.2、生产者同步或异步发送消息
 * 5.3、生产者消息发送重试机制：保证消息不丢失
 * 5.4、生产者幂等性：解决重试机制消息重复&乱序
 * 5.5、生产者事务
 * @Author Submerge.
 * @Since 2025/5/31 00:50
 * @Version 1.0
 */
public class SubKafkaProducer {

    private static KafkaProducer<String, String> producer;

    public static void main(String[] args) throws Exception{

        // TODO 配置属性集合
        Map<String, Object> configMap = new HashMap<>();
        // TODO 配置属性：Kafka服务器集群地址
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // TODO 配置属性：Kafka生产的数据为KV对，所以在生产数据进行传输前需要分别对K,V进行对应的序列化操作
        configMap.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        configMap.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        //TODO 配置自定义拦截器 @link {SubProducerValueInterceptor}
        //configMap.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, SubProducerValueInterceptor.class.getName());

        //TODO 配置分区器 @link {SubPartitioner}
        //configMap.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, SubPartitioner.class.getName());

        //TODO 配置ACK确认机制，可以配置为0,1,all（等同于-1）
        // 分别表示：0-不确认（即发送成功即可）; 1-确认leader节点; all-确认所有节点，默认为all
        configMap.put(ProducerConfig.ACKS_CONFIG, "all");

        //TODO 失败重试次数
        configMap.put(ProducerConfig.RETRIES_CONFIG, 3);

        //TODO
        //configMap.put(ProducerConfig.BATCH_SIZE_CONFIG, 5);

        //TODO 启用幂等性
        //configMap.pu每个连接的在途请求数，不能大于5，取值范围为[1,5]t(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        //每个连接的在途请求数，不能大于5，取值范围为[1,5]
        //configMap.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);

        //TODO 事务，事务依赖幂等性
        //configMap.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "sub-transactional-id");
        //configMap.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 5000);
        //事务操作如下，分别为初始、开启、提交、终止（回滚）
        //producer.initTransactions();
        //producer.beginTransaction();
        //producer.commitTransaction();
        //producer.abortTransaction();


        // TODO 创建Kafka生产者对象，建立Kafka连接 (构造对象时，需要传递配置参数)
        producer = new KafkaProducer<>(configMap);



        // TODO 准备数据,定义泛型 (构造对象时需要传递 【Topic主题名称】，【Key】，【Value】三个参数)
        for (int i = 1; i <= 100; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(
                    "sub-spark-kafka", "hello" + i, "spark" + i
            );

            // TODO 生产（发送）数据
            producer.send(record).get();
            Thread.sleep(1000);
        }


        //异步发送-默认即为异步
        //sendAsync(record);

        //同步发送-会阻塞当前线程，直到发送成功
        //sendSync(record);



        // TODO 关闭生产者连接
        producer.close();


    }

    /**
     *  异步发送
     */
    public static void sendAsync(ProducerRecord<String, String> record) {
        //异步发送
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("发送成功：" + metadata);
            } else {
                System.out.println("发送失败：" + exception);
            }
        });
        System.out.println("异步发送完成");
    }

    /**
     *  同步发送，会阻塞当前线程，直到发送成功
     *  这里使用JUC包中的Future
     */
    public static void sendSync(ProducerRecord<String, String> record) throws ExecutionException, InterruptedException {
        Future<RecordMetadata> future = producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("发送成功：" + metadata);
            } else {
                System.out.println("发送失败：" + exception);
            }
        });

        //同步发送，会阻塞当前线程，直到发送成功
        future.get();
        System.out.println("同步发送完成");
    }

}
